package com.iussoft.receiveData.kafka;

import com.iussoft.kafka.KafkaConfig;
import com.iussoft.kafka.KakfaService;
import com.iussoft.receiveData.schedul.ScheduledDynamicTask;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.core.env.Environment;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * @description: 定义kafka消息生产者
 * @author: Hou Dayu
 * @date: Created in 2020/3/1
 */
@Component
public class KafkaSimpleSender {

    @Autowired(required = false)
    @Qualifier("kafkaConnector")
    private KakfaService KakfaService;

    @Autowired
    private KafkaConfig kafkaConfig;

    @Autowired
    private Environment env;

    @Autowired
    @Qualifier("scheduledDynamicTask")
    private ScheduledDynamicTask dynamicTask;

    /**
     * 消费者并发消息监听容器
     */
    public void init() {
        if(KakfaService.getKakfaTemplate() == null) {
            KakfaService.setKafkaConfig(kafkaConfig);
        }
    }

    /**
     * 发送消息
     * @param topic Topic的名字
     * @param key：消息的键
     * @param data 消息的数据
     */
    public void sendMsg(final String topic, final String key, final Object data){
        if("1".equals(kafkaConfig.getEnabled())) {
            init();
            dynamicTask.submitLog(new Runnable() {//异步发送
                @Override
                public void run() {
                    try {
                        ListenableFuture<SendResult<String, Object>> future = KakfaService.sendMessage(topic, 0, key, data);
                        future.addCallback(new KafkaMsgSuccess(), new KafakMsgCallback());
                    }catch (Exception e){}
                }
            });
        }
    }

    /**
     * 消费者并发消息监听容器(接入报文监听)
     */
    @Bean(name="receiveMessageListenerContainer", initMethod = "doStart")
    public ConcurrentMessageListenerContainer concurrentMessageListenerContainer(){
        if("1".equals(kafkaConfig.getEnabled())) {
            String topic = env.getProperty("kafka.receive.topic");
            String[] topics = topic.split(",");
            ContainerProperties properties = new ContainerProperties(topics);
            int concurrency = Integer.valueOf(kafkaConfig.getConcurrency());
            ConcurrentMessageListenerContainer listenerContainer = KakfaService.initConsumerConfig(kafkaConfig, new KafakMessageServcie(topic,dynamicTask), concurrency, properties);
            return listenerContainer;
        }
        return null;
    }

    /**
     * 消费者并发消息监听容器(异步报文监听)
     */
    @Bean(name="asyncMessageListenerContainer", initMethod = "doStart")
    public ConcurrentMessageListenerContainer asyncMessageListenerContainer(){
        if("1".equals(kafkaConfig.getEnabled())) {
            String topic = env.getProperty("kafka.send.async.topic");
            String[] topics = topic.split(",");
            ContainerProperties properties = new ContainerProperties(topics);
            int concurrency = Integer.valueOf(kafkaConfig.getConcurrency());
            ConcurrentMessageListenerContainer listenerContainer = KakfaService.initConsumerConfig(kafkaConfig, new AsyncKafakMessageServcie(topic), concurrency, properties);
            return listenerContainer;
        }
        return null;
    }

    public boolean getKafkaEnabled(){
        return "1".equals(kafkaConfig.getEnabled());
    }
}
